package edu.buaa.traffic.client;


import com.google.common.util.concurrent.ListenableFuture;
import edu.buaa.common.client.DBProxy;
import edu.buaa.common.client.PostgreSqlExecutorClient;
import edu.buaa.common.transaction.AbstractTransaction;
import edu.buaa.traffic.model.StatusUpdate;
import edu.buaa.traffic.transaction.*;
import edu.buaa.traffic.transaction.internal.NodeNeighborRoadTx;
import edu.buaa.traffic.algo.GeneralizedDijkstra;
import edu.buaa.traffic.algo.Topology;
import edu.buaa.utils.Pair;
import edu.buaa.utils.Triple;

import java.sql.*;
import java.util.*;
import java.util.concurrent.*;

public class PostgreSQLExecutorClient extends PostgreSqlExecutorClient {
    private final HashMap<String, Long> name2Id = new HashMap<>();
    private final HashMap<Long, String> id2Name = new HashMap<>();

    public PostgreSQLExecutorClient(String serverHost, int parallelCnt) throws Exception {
        super(serverHost, "traffic", parallelCnt);
    }

    @Override
    protected List<String> createTables() {
        return Arrays.asList(
                "create table IF NOT EXISTS road(r_id bigint primary key, r_address char(12), r_start bigint, r_end bigint, r_length int, r_type int)",
                "create table IF NOT EXISTS cross_node(cross_id bigint primary key, cross_name text)",
                "create table IF NOT EXISTS temporal_status(ts_id serial not null, t int, r_id bigint, status int, travel_t int, seg_cnt int)",
                "create index IF NOT EXISTS ind_id_t on temporal_status(r_id, t)"
        );
    }

    @Override
    public ListenableFuture<ServerResponse> execute(AbstractTransaction tx) throws InterruptedException {
        switch (tx.getTxType()) {
            case tx_import_static_data:
                return this.submit(execute((ImportStaticDataTx) tx));
            case tx_import_temporal_data:
                return this.submit(execute((ImportTemporalDataTx) tx), tx.getSection());
            case tx_query_road_by_temporal_condition:
                return this.submit(execute((EntityTemporalConditionTx) tx));
            case tx_query_node_neighbor_road:
                return this.submit(execute((NodeNeighborRoadTx) tx));
            case tx_query_snapshot:
                return this.submit(execute((SnapshotQueryTx) tx));
            case tx_update_temporal_data:
                return this.submit(execute((UpdateTemporalDataTx) tx));
            case tx_query_snapshot_aggr_max:
                return this.submit(execute((SnapshotAggrMaxTx) tx));
            case tx_index_tgraph_aggr_duration:
                return this.submit(execute((SnapshotAggrDurationTx) tx));
            case tx_query_reachable_area:
                return this.submit(execute((ReachableAreaQueryTx) tx));
            default:
                throw new UnsupportedOperationException();
        }
    }

    @Override
    public String testServerClientCompatibility() {
        return null;
    }

    protected void connected(Connection conn) throws Exception {
        super.connected(conn);
        buildMapBetweenIdAndName(conn);
    }

    private void buildMapBetweenIdAndName(Connection conn) throws SQLException {
        try (Statement stmt0 = conn.createStatement()){
            ResultSet rs = stmt0.executeQuery("select r_id, r_address from road");
            while (rs.next()) {
                Long id = rs.getLong("r_id");
                String name = rs.getString("r_address");
                id2Name.put(id, name);
                name2Id.put(name, id);
            }
        }
    }

    private Req execute(ImportStaticDataTx tx) {
        return conn -> {
            try (PreparedStatement stmt0 = conn.prepareStatement("insert into cross_node values(?, ?)");
                 PreparedStatement stmt1 = conn.prepareStatement("insert into road values(?, ?, ?, ?, ?, ?)")) {
                conn.setAutoCommit(false);
                for (ImportStaticDataTx.StaticCrossNode cross : tx.getCrosses()) {
                    stmt0.setLong(1, cross.getId());
                    stmt0.setString(2, cross.getName());
                    stmt0.addBatch();
                }
                stmt0.executeBatch();
                for (ImportStaticDataTx.StaticRoadRel road : tx.getRoads()) {
                    stmt1.setLong(1, road.getRoadId());
                    stmt1.setString(2, road.getId());
                    stmt1.setLong(3, road.getStartCrossId());
                    stmt1.setLong(4, road.getEndCrossId());
                    stmt1.setInt(5, road.getLength());
                    stmt1.setInt(6, road.getType());
                    stmt1.addBatch();
                }
                stmt1.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            buildMapBetweenIdAndName(conn);
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(ImportTemporalDataTx tx) {
        return conn -> {
            try (PreparedStatement stmt = conn.prepareStatement("insert into temporal_status(r_id, t, status, travel_t, seg_cnt) values(?, ?, ?, ?, ?)")) {
                conn.setAutoCommit(false);
                for (StatusUpdate s : tx.data) {
                    Long id = name2Id.get(s.getRoadId());
                    if (id == null) {
                        System.out.println("Strange case occurred:");
                        System.out.println("name " + s.getRoadId() + " did not be present at road table.");
                    }
                    stmt.setLong(1, id);
                    stmt.setInt(2, s.getTime());
                    stmt.setInt(3, s.getJamStatus());
                    stmt.setInt(4, s.getTravelTime());
                    stmt.setInt(5, s.getSegmentCount());
                    stmt.addBatch();
                }
                stmt.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(UpdateTemporalDataTx tx) {
        return conn -> {
            conn.setAutoCommit(false);
            Long roadId = name2Id.get(tx.getRoadId());
            try (PreparedStatement stmt1 = conn.prepareStatement("select r_id, max(t) as max_t from temporal_status where r_id = ? and t <= ? group by r_id");
                 PreparedStatement stmt2 = conn.prepareStatement("select status, travel_t, seg_cnt from temporal_status where r_id = ? and t = ?");
                 PreparedStatement stmt3 = conn.prepareStatement("insert into temporal_status(t, r_id, status, travel_t, seg_cnt) values(?, ?, ?, ?, ?)");
                 PreparedStatement stmt4 = conn.prepareStatement("delete from temporal_status where r_id = ? and t >= ? and t <= ?")) {
                stmt1.setLong(1, roadId);
                stmt1.setInt(2, tx.getEndTime());
                ResultSet rs = stmt1.executeQuery();

                // insert into table from t1 + 1 with the old value.
                if (rs.next()) {
                    stmt2.setLong(1, rs.getLong("r_id"));
                    stmt2.setInt(2, rs.getInt("max_t"));
                    ResultSet r = stmt2.executeQuery();
                    if (r.next()) {
                        stmt3.setInt(1, tx.getEndTime() + 1);
                        stmt3.setLong(2, roadId);
                        stmt3.setInt(3, r.getInt("status"));
                        stmt3.setInt(4, r.getInt("travel_t"));
                        stmt3.setInt(5, r.getInt("seg_cnt"));
                        stmt3.execute();
                    }
                }
                // delete the [t0, t1] rows.
                stmt4.setLong(1, roadId);
                stmt4.setInt(2, tx.getStartTime());
                stmt4.setInt(3, tx.getEndTime());
                stmt4.execute();
                // insert into table from t0 with the new value.
                stmt3.setInt(1, tx.getStartTime());
                stmt3.setLong(2, roadId);
                stmt3.setInt(3, tx.getJamStatus());
                stmt3.setInt(4, tx.getTravelTime());
                stmt3.setInt(5, tx.getSegmentCount());
                stmt3.execute();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(SnapshotQueryTx tx) {
        return conn -> {
            List<Pair<String, Integer>> res = new ArrayList<>();
            try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, max(t) as max_t from temporal_status where t <= ? group by r_id");
                 PreparedStatement stmt1 = conn.prepareStatement("select " + tx.getPropertyName() + " as property from temporal_status where r_id = ? and t = ?")) {
                conn.setAutoCommit(false);
                int snapshotTime = tx.getTimestamp();
                stmt0.setInt(1, snapshotTime);
                ResultSet rs0 = stmt0.executeQuery();
                while (rs0.next()) {
                    stmt1.setLong(1, rs0.getLong("r_id"));
                    stmt1.setInt(2, rs0.getInt("max_t"));
                    ResultSet rs1 = stmt1.executeQuery();
                    if (rs1.next()) {
                        res.add(Pair.of(id2Name.get(rs0.getLong("r_id")), rs1.getInt("property")));
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            SnapshotQueryTx.Result result = new SnapshotQueryTx.Result();
            result.setRoadStatus(res);
            return result;
        };
    }

    private Req execute(SnapshotAggrMaxTx tx) {
        return conn -> {
            List<Pair<String, Integer>> res = new ArrayList<>();
            try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, max(t) as real_st from temporal_status where t <= ? group by r_id");
                 PreparedStatement stmt2 = conn.prepareStatement("select max(" + tx.getP() + ") as max_p from temporal_status where r_id = ? and t >= ? and t <= ? group by r_id")) {
                conn.setAutoCommit(false);
                stmt0.setInt(1, tx.getT0());
                ResultSet rs = stmt0.executeQuery();
                HashMap<String, Integer> st = new HashMap<>();
                while (rs.next()) {
                    Long id = rs.getLong("r_id");
                    stmt2.setLong(1, id);
                    stmt2.setInt(2, rs.getInt("real_st"));
                    stmt2.setInt(3, tx.getT1());
                    ResultSet r = stmt2.executeQuery();
                    while (r.next()) {
                        res.add(Pair.of(id2Name.get(id), r.getInt("max_p")));
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            SnapshotAggrMaxTx.Result result = new SnapshotAggrMaxTx.Result();
            result.setRoadTravelTime(res);
            return result;
        };
    }

    private Req execute(SnapshotAggrDurationTx tx) {
        return conn -> {
            List<Triple<String, Integer, Integer>> res = new ArrayList<>();
            try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, max(t) as real_st from temporal_status where t <= ? group by r_id");
                 PreparedStatement stmt1 = conn.prepareStatement("select t, " + tx.getP() + " as property from temporal_status where r_id = ? and t >= ? and t <= ? order by t")) {
                conn.setAutoCommit(false);
                stmt0.setInt(1, tx.getT0());
                ResultSet rs = stmt0.executeQuery();
                while (rs.next()) {
                    Long id = rs.getLong("r_id");
                    stmt1.setLong(1, id);
                    stmt1.setInt(2, rs.getInt("real_st"));
                    stmt1.setInt(3, tx.getT1());
                    ResultSet r = stmt1.executeQuery();
                    int lastTime = -1, lastStatus = -1;
                    HashMap<Integer, Integer> buffer = new HashMap<>();
                    while (r.next()) {
                        int property = r.getInt("property");
                        int t = r.getInt("t");
                        if (lastTime != -1) {
                            int duration = t - lastTime;
                            buffer.merge(lastStatus, duration, Integer::sum);
                            lastTime = t;
                        } else {
                            lastTime = tx.getT0();
                        }
                        lastStatus = property;
                    }
                    if (lastTime != -1) buffer.merge(lastStatus, tx.getT1() - lastTime, Integer::sum);
                    for (Map.Entry<Integer, Integer> e : buffer.entrySet()) {
                        res.add(Triple.of(id2Name.get(id), e.getKey(), e.getValue()));
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            SnapshotAggrDurationTx.Result result = new SnapshotAggrDurationTx.Result();
            result.setRoadStatDuration(res);
            return result;
        };
    }

    private Req execute(EntityTemporalConditionTx tx) {
        return conn -> {
            List<String> res = new ArrayList<>();
            try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, max(t) as real_st from temporal_status where t <= " + tx.getT0() + " group by r_id");
                 PreparedStatement stmt1 = conn.prepareStatement("select r_id from temporal_status where r_id = ? and t >= ? and t <= ? and " + tx.getP() + " >= ? and " + tx.getP() + " <= ? limit 1")) {
                conn.setAutoCommit(false);
                ResultSet rs = stmt0.executeQuery();
                while (rs.next()) {
                    Long id = rs.getLong("r_id");
                    stmt1.setLong(1, id);
                    stmt1.setInt(2, rs.getInt("real_st"));
                    stmt1.setInt(3, tx.getT1());
                    stmt1.setInt(4, tx.getVmin());
                    stmt1.setInt(5, tx.getVmax());
                    ResultSet r = stmt1.executeQuery();
                    if (r.next()) {
                        res.add(id2Name.get(id));
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            EntityTemporalConditionTx.Result result = new EntityTemporalConditionTx.Result();
            result.setRoads(res);
            return result;
        };
    }

    private Req execute(NodeNeighborRoadTx tx) {
        return conn -> {
            List<String> answer = new ArrayList<>();
            try (PreparedStatement stmt = conn.prepareStatement("select r_name from road where r_start = " + tx.getNodeId() + " or r_end = " + tx.getNodeId())) {
                ResultSet rs = stmt.executeQuery();
                while (rs.next()) {
                    answer.add(rs.getString("r_name"));
                }
            }
            NodeNeighborRoadTx.Result result = new NodeNeighborRoadTx.Result();
            result.setRoadIds(answer);
            return result;
        };
    }


    private Req execute(ReachableAreaQueryTx tx) {
        return conn -> GeneralizedDijkstra.solve(tx, new PostgreSQLTopology(conn));
    }

    private static class PostgreSQLTopology implements Topology {

        private final Connection conn;

        PostgreSQLTopology(Connection conn) {
            this.conn = conn;
        }

        @Override
        public ArrayList<Pair<Long, Long>> getRoadList(long curCross) {
            ArrayList<Pair<Long, Long>> ret = new ArrayList<>();
            try (PreparedStatement stmt = conn.prepareStatement("select r_id, r_end from road where r_start = ?")) {
                stmt.setLong(1, curCross);
                ResultSet rs = stmt.executeQuery();
                while (rs.next())
                    ret.add(Pair.of(rs.getLong("r_id"), rs.getLong("r_end")));
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return ret;
        }

        @Override
        public int earliestArriveTime(long roadId, int when, int until) {
            int ret = Integer.MAX_VALUE;
            try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, max(t) as max_t from temporal_status where r_id = ? and t <= ? group by r_id");
                 PreparedStatement stmt1 = conn.prepareStatement("select r_id, min(t) as min_t from temporal_status where r_id = ? and t >= ? group by r_id");
                 PreparedStatement stmt2 = conn.prepareStatement("select t, travel_t from temporal_status where r_id = ? and t >= ? and t <= ?")) {
                stmt0.setLong(1, roadId);
                stmt0.setInt(2, when);
                ResultSet rs = stmt0.executeQuery();
                int st = -1;
                if (rs.next()) st = rs.getInt("max_t");
                stmt1.setLong(1, roadId);
                stmt1.setInt(2, until);
                rs = stmt1.executeQuery();
                int en = Integer.MAX_VALUE;
                if (rs.next()) en = rs.getInt("min_t");
                stmt2.setLong(1, roadId);
                stmt2.setInt(2, st);
                stmt2.setInt(3, en);
                rs = stmt2.executeQuery();
                ArrayList<Pair<Integer, Integer>> l = new ArrayList<>();
                while (rs.next()) l.add(Pair.of(rs.getInt("t"), rs.getInt("travel_t")));
                if (en == Integer.MAX_VALUE) l.add(Pair.of(Integer.MAX_VALUE, 0));
                l.sort(Comparator.comparingInt(Pair::getKey));
                int sz = l.size();
                if (sz == 1) return when + l.get(0).getValue();
                for (int i = 0; i < sz - 1; ++i) {
                    if (i != 0) when = l.get(i).getKey();
                    int cur = when + l.get(i).getValue();
                    int next = l.get(i + 1).getKey();
                    if (cur <= next && cur <= until && cur < ret) ret = cur;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return ret;
        }
    }

}
